package com.feihao.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.feihao.dao.AccountDao;
import com.feihao.domian.AccountEvent;
import com.feihao.service.AccountService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.ObjectUtils;

import java.math.BigDecimal;
import java.nio.charset.StandardCharsets;
import java.util.concurrent.TimeUnit;

@Slf4j
@Component
public class AccountServiceImpl implements AccountService {

    @Autowired
    private AccountService accountService;

    @Autowired
    private AccountDao accountDao;
    public static TransactionMQProducer producer;

    static{
         producer = new TransactionMQProducer("group1");
        //事务消息：
        producer.setSendMsgTimeout(10000);

        producer.setNamesrvAddr("101.34.13.22:9876");

        //开启服务：
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    @Override
    public void sendUpdateAccountBalance(AccountEvent ace) throws MQClientException {
        //构建消息体：
        JSONObject jsonObject = new JSONObject();
        jsonObject.put("accountChange",ace);


//        producer = new TransactionMQProducer("group1");
//        //事务消息：
//        producer.setSendMsgTimeout(10000);
//
//        producer.setNamesrvAddr("101.34.13.22:9876");

        producer.setTransactionListener(new TransactionListener() {
            //正常事务过程：
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                try {
                    //解析消息内容：
                    String jsonString = new String(message.getBody());
                    JSONObject jsonObject1 = JSONObject.parseObject(jsonString);
                    AccountEvent ace = JSONObject.parseObject(jsonObject1.getString("accountChange"), AccountEvent.class);

                    //调用service的方法:调整金额：
                    boolean b = accountService.doUpdateAccountBalance(ace);
                    if (b){  //成功修改：
                        //TimeUnit.SECONDS.sleep(100);
                        return LocalTransactionState.UNKNOW;
                    }else {
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                } catch (Exception e) {
                    System.out.println("事务执行失败！");
                    e.printStackTrace();
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }


            //事务补偿：
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                final JSONObject jsonObject = JSON.parseObject(new String(messageExt.getBody()));
                AccountEvent ace = JSONObject.parseObject(jsonObject.getString("accountChange"), AccountEvent.class);
                //事务id
                String txNo = ace.getTxNo();
                int existTxA = accountDao.isExistTxA(txNo);
                log.info("回查事务号：{}，结果：{}",ace.getTxNo(),existTxA);

                if (existTxA >0){
                    return LocalTransactionState.COMMIT_MESSAGE;
                }else {
                    return LocalTransactionState.ROLLBACK_MESSAGE;
                }
            }
        });

//        producer.start();

        Message message = new Message("accountTxTopic",jsonObject.toJSONString().getBytes(StandardCharsets.UTF_8));
        TransactionSendResult sendResult = producer.sendMessageInTransaction(message, null);
        System.out.println("消息体：" +new String(message.getBody()) + "发送结果：" +sendResult.getSendStatus() );

        try {
            TimeUnit.SECONDS.sleep(2);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        //事务补偿必须保证服务运行，不能shutdown!
//        producer.shutdown();

    }

    @Override
    public boolean doUpdateAccountBalance(AccountEvent ace) {
        System.out.println("开始更新本地事务，事务号："+ ace.getTxNo());

        //查询余额是否足够：
        Double totalMoney = accountDao.queryAccountBalance(ace.getAccountNoA());

        int flag = (int) (totalMoney * 100 - (ace.getAmount() * 100));

        //如果查询的余额为空  获取余额减去实际要发送的金额 的差小于0
        if (ObjectUtils.isEmpty(totalMoney) || flag <0){
            return false;
        }

        accountDao.subtractAccountBalance(ace.getAccountNoA(),ace.getAmount());

        //为幂等性做准备：
        int i = accountDao.addTxA(ace.getTxNo());


        //测试
        /*if (ace.getAmount().compareTo(0.0) == 0) {
            throw new RuntimeException("bank1更新本地事务时抛出异常");
        }
        log.info("结束更新本地事务，事务号：{}", ace.getTxNo());*/

        if (i>0){
            return true;
        }else {
            return false;
        }
    }
}
